package org.example;

import lombok.SneakyThrows;

import javax.swing.plaf.synth.SynthOptionPaneUI;
import java.io.*;
import java.net.*;
import java.text.DecimalFormat;



/*
 * 客户端
 */
public class UDPClient  implements Runnable{//公共类



    private InetSocketAddress socketAddress;//发送IP端口设置

    //发送的数据编码
    byte[] data1;
    //接收到后处理好的数据
    String[] datalist;

    private boolean txt;

    private boolean producer;

    private boolean consumerToMysql;
    //接收socket套接字
    private  DatagramSocket socket;
    //消费到哪个分区
    private int partition;

    //消费者
    Producer producer1 = new Producer();

    UDPClient(DatagramSocket receiveSocket,String serverIp,boolean txt,boolean producer,boolean consumerToMysql){

        this.producer=producer;
        this.consumerToMysql=consumerToMysql;
        this.txt=txt;
        this.socket = receiveSocket;


        /*
         * 向服务器端发送数据
         */

        datalist = new String[6];//创建double数组来存储数据

        //1.创建数据
        byte[] data = new byte[1024];
        data[0] = (byte)0x01;
        data[1] = (byte)0x03;
        data[2] = (byte)0x10;
        data[3] = (byte)0x00;
        data[4] = (byte)0x00;
        data[5] = (byte)0x0A;
        data[6] = (byte)0xF0;
        data[7] = (byte)0x7B;
//        System.out.println(data);

        //1.创建数据
        String str ="01 03 10 00 00 0A C1 0D";
//        String str ="01 03 10 00 00 0A F0 7B";
        data1= hexStringToByteArray(str);

        //发送到指定IP端口的数据包属性初始化
        socketAddress = new InetSocketAddress(serverIp,608);




    }

    @SneakyThrows
    public void start(boolean txt,boolean producer,boolean consumerToMysql){

        /*
         * 是否开启consumer(先保证消费者打开)
         * 需要创建线程，否则阻塞不能向下执行
         * */
        Runnable ra = new Runnable() {
            @Override
            public void run() {
                if(consumerToMysql){
                    writeToMysql();
                }
            }
        };
        Thread thread = new Thread(ra);
        thread.start();


        DatagramPacket packet = new DatagramPacket(data1, data1.length,socketAddress);//2.创建数据报，包含发送的数据信息
//        byte[] data2 = new byte[25];//创建字节数组(此处接收的数据格式为固定长度,所以创建了固定大小的字节数组)
//        DatagramPacket packet2 = new DatagramPacket(data2, data2.length,socketAddress);// 1.创建数据报，用于接收服务器端响应的数据

        /*
         * 发送给上位机数据码
         * */

        DatagramSocket socket = this.socket; // 3.创建DatagramSocket对象


        DecimalFormat df = new DecimalFormat("#,#00.00#");//指定输出格式
        String errDevice= null;
        int errorFlag=0;
        int time = 1;
        while(true) {//通过循环不同的向客户端发送和接受数据
            //将以下两行放在此处，是为了防止接收不到数据时，写入之前所获取的数据，即每次获取数据时，将数据初始化。
            byte[] data2 = new byte[25];//创建字节数组(此处接收的数据格式为固定长度,所以创建了固定大小的字节数组)
            DatagramPacket packet2 = new DatagramPacket(data2, data2.length,socketAddress);// 1.创建数据报，用于接收服务器端响应的数据
//            System.out.println("返回上位机之前");
            try {
                errDevice=packet.getAddress().getHostAddress();
//                System.out.println(errDevice);
//               data2 = new byte[25];
                socket.send(packet);// 4.向上位机发送数据报
            } catch (IOException e) {
                System.out.println("发送出错");

            }

//            System.out.println("接收上位机之前");
            /*
             * 接收服务器端响应的数据
             */
//            while(true) {
                try {
                    //设置超时时长,这样就可以防止发生阻塞，导致程序不运行
                    socket.setSoTimeout(200);
                    socket.receive(packet2);// 2.接收上位机响应的数据
                } catch (IOException e) {
                    Thread.sleep(800);//休眠1秒
                    System.out.println("某个设备接收出错!!!!!!!!!!!!!!!!!!!!!!");
                    continue;
                }
//            System.out.println("a接收到上位机响应的数据a");
//            }

//            System.out.println(packet2.getAddress().getHostAddress());
                if(packet2.getAddress().getHostAddress().equals("192.168.100.151")){
                    this.partition=0;
                } else if (packet2.getAddress().getHostAddress().equals("192.168.100.152")) {
                    this.partition=1;
                } else if (packet2.getAddress().getHostAddress().equals("192.168.100.153")) {
                    this.partition=2;
                }
//
//            while
//                如果false ，继续获取，同时弹出提示
//                返回值=true


            //获取数据
            //3.读取数据
            String reply = BinaryToHexString(data2);
            System.out.println("我是客户端"+this.partition+"，上位机说：" + reply);//输出提示信息

            /*
             * 数据处理
             * */
            //去掉空格
            String replaceAll = reply.replaceAll(" ", "");


            //TODO 读取特定位置的字符并转换成温度和湿度
            //截取其中的6个数字
            int i=0;
            int j=6;//定长截取字符串



            while (i<6){
                String substring = replaceAll.substring(j, j + 4);//截取各个数据
                Long parseLong = Long.parseLong(substring, 16);//16进制转换10进制
//                System.out.println(parseLong);
//                double d = Double.parseDouble(df.format(parseLong/100.00));//long转换为double
//                System.out.println(df.format(parseLong/100.00));
                if(i==5){
                    datalist[i]=String.format("%.1f",(parseLong/100.0));//数据写入到数组中
                    System.out.println(String.format("%.1f",(parseLong/100.0)));
                }else {
                    datalist[i]=df.format(parseLong/100.00);//数据写入到数组中
                    System.out.println(df.format(parseLong/100.00));
                }


                j+=4;
                i++;
            }

            /*
            * 是否写入txt
            * */
            if (txt){
                writeToTxt(datalist);
            }
            /*
            * 是否开始producer
            * */
            if(producer){
//                System.out.println("写入kafka前");
                writeToKafka(datalist);
            }
            System.out.println("写入kafka后");

            System.out.println("--------------------------------------");
            System.out.println(time);

            Thread.sleep(1000);//休眠1秒

            time++;
        }
        //改为true之后此处不会执行
//        socket.close();//4.关闭资源


    }

    @SneakyThrows
    private void writeToTxt(String[] datalist){
        /*
         * 将一组数据写入txt文件中
         * */
        String path = "C:\\Users\\57192\\Desktop\\11.txt";
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(path,true)));//可以追加

        for (String d: datalist){
            bw.write(String.valueOf(d));
            bw.write(" ");
        }
        bw.newLine();
        bw.close();

    }

    private void writeToKafka(String[] datalist){
        /*
         * 产生数据到kafka
         * */

        producer1.send(datalist,this.partition);
        System.out.println("分区"+this.partition+"写入数据");

    }

    private void writeToMysql(){
        /*
        * 消费数据到Mysql
        * */
        Consumer consumer = new Consumer("temp");
        consumer.mysql_consume();

    }

    //string to byte[]
    public static byte[] hexStringToByteArray(String hexString) {
        hexString = hexString.replaceAll(" ", "");
        int len = hexString.length();
        byte[] bytes = new byte[len / 2];
        for (int i = 0; i < len; i += 2) {
            // 两位一组，表示一个字节,把这样表示的16进制字符串，还原成一个字节
            bytes[i / 2] = (byte) ((Character.digit(hexString.charAt(i), 16) << 4) + Character
                    .digit(hexString.charAt(i + 1), 16));
        }
        return bytes;
    }

    //byte[] to string
    public static String BinaryToHexString(byte[] bytes) {
        String hexStr = "0123456789ABCDEF";
        String result = "";
        String hex = "";
        for (byte b : bytes) {
            hex = String.valueOf(hexStr.charAt((b & 0xF0) >> 4));
            hex += String.valueOf(hexStr.charAt(b & 0x0F));
            result += hex + " ";
        }
        return result;
    }

    /*
     *
     * CRC16/Modbus冗余校验
     * */
    public static String getCRC2(byte[] bytes) {
//        ModBus 通信协议的 CRC ( 冗余循环校验码含2个字节, 即 16 位二进制数。
//        CRC 码由发送设备计算, 放置于所发送信息帧的尾部。
//        接收信息设备再重新计算所接收信息 (除 CRC 之外的部分）的 CRC,
//        比较计算得到的 CRC 是否与接收到CRC相符, 如果两者不相符, 则认为数据出错。
//
//        1) 预置 1 个 16 位的寄存器为十六进制FFFF(即全为 1) , 称此寄存器为 CRC寄存器。
//        2) 把第一个 8 位二进制数据 (通信信息帧的第一个字节) 与 16 位的 CRC寄存器的低 8 位相异或, 把结果放于 CRC寄存器。
//        3) 把 CRC 寄存器的内容右移一位( 朝低位)用 0 填补最高位, 并检查右移后的移出位。
//        4) 如果移出位为 0, 重复第 3 步 ( 再次右移一位); 如果移出位为 1, CRC 寄存器与多项式A001 ( 1010 0000 0000 0001) 进行异或。
//        5) 重复步骤 3 和步骤 4, 直到右移 8 次,这样整个8位数据全部进行了处理。
//        6) 重复步骤 2 到步骤 5, 进行通信信息帧下一个字节的处理。
//        7) 将该通信信息帧所有字节按上述步骤计算完成后,得到的16位CRC寄存器的高、低字节进行交换。
//        8) 最后得到的 CRC寄存器内容即为 CRC码。

        int CRC = 0x0000ffff;
        int POLYNOMIAL = 0x0000a001;

        int i, j;
        for (i = 0; i < bytes.length; i++) {
            CRC ^= (int) bytes[i];
            for (j = 0; j < 8; j++) {
                if ((CRC & 0x00000001) == 1) {
                    CRC >>= 1;
                    CRC ^= POLYNOMIAL;
                } else {
                    CRC >>= 1;
                }
            }
        }
        //高低位转换
        CRC = ( (CRC & 0x0000FF00) >> 8) | ( (CRC & 0x000000FF ) << 8);
        return Integer.toHexString(CRC);
    }
/*
* 重写方法
*
* */

    @Override
    public void run() {
        System.out.println("线程启动！");
        this.start(this.txt,this.producer,this.consumerToMysql);

    }
}